package cn.edu.laud.sany.datatest;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

import cn.edu.laud.sany.types.ImportProcess;
import cn.edu.laud.sany.types.RowData;
import cn.edu.laud.sany.types.TimeSerial;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;

public class LaUDTest_CSVReader extends Thread
{
	protected MyLog logger;
	protected SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	protected SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
	// 当前处理的时间，用来做标志以确定如何添加流水号
	protected Map<String, TimeSerial> currenttime = new HashMap<String, TimeSerial>();
	// 流水号
	protected TimeSerial timeSerial = null;
	
	protected List<RowData> rowList = new ArrayList<RowData>();
	protected List<String> sqlList = new ArrayList<String>();
	private ConcurrentLinkedQueue<String> sqlQueue;
	protected int threadNum;
	
	private File file = null;
	private long cycleCount = 0;
	private BufferedReader bReader;
	private long nowLineNum;
	private String fileName;
	
	public LaUDTest_CSVReader(int no)
	{
		this.sqlQueue = LaUDTest_JDBC.sqlQueue;
		this.threadNum = no;
		logger = MyLog.getMyLog(LaUDImportTest.getLogFilePath(), true);
		logger.setThreadName("Thread_Reader " + no);
	}
	
	private void makeSQL(int paraId, String equipCode, String paravalue, long localTime)
	{
		sqlList.add(String.format(LaUDImportTest.cQLInsertFormat, paraId, localTime, equipCode,
				paravalue));
	}
	
	private String trimData(String oriData)
	{
		oriData = oriData.trim();
		oriData = oriData.substring(1, oriData.length() - 1);
		return oriData;
	}
	
	protected RowData getRowData(String string)
	{
		String data[] = null;
		String equipCode = null;
		int paraId = 0;
		String paravalue = null;
		long localTime = 0L;
		try
		{
			data = string.split(",");
			if (data.length != 6)
			{
				logger.error("data format error.(org:)" + string);
				System.err.println("data format error.(org:)" + string);
				return null;
			}
			equipCode = trimData(data[1]);
			paraId = Integer.valueOf(trimData(data[2]));
			if (paraId == 356)
			{
				return null;
			}
			paravalue = trimData(data[3]);
			String dataString = trimData(data[5]);
			try
			{
				localTime = sdf.parse(dataString).getTime();
			}
			catch (ParseException e)
			{
				try
				{
					localTime = sdf2.parse(dataString).getTime();
				}
				catch (ParseException e1)
				{
					e1.printStackTrace();
					logger.error(e1.getMessage() + "(ori:)" + string);
					
					return null;
				}
			}
		}
		catch (NumberFormatException e1)
		{
			e1.printStackTrace();
			logger.error(e1.getMessage() + "(ori:)" + string);
			return null;
		}
		// 添加流水号
		timeSerial = currenttime.get(equipCode + paraId);
		if (timeSerial == null)
		{
			timeSerial = new TimeSerial(localTime, 0);
			currenttime.put(equipCode + paraId, timeSerial);
		}
		else
		{
			if (localTime != timeSerial.time)
			{
				timeSerial.time = localTime;
				timeSerial.serial = 0;
			}
			else
			{
				timeSerial.serial++;
				localTime += timeSerial.serial;
			}
		}
		RowData rowData = new RowData();
		rowData.paraId = paraId;
		rowData.paravalue = paravalue;
		rowData.equipCode = equipCode;
		rowData.localTime = localTime;
		return rowData;
	}
	
	protected boolean readData_SQL()
	{
		boolean finishFile = true;
		if (nowLineNum == 0)
		{
			logger.info("find a new input file: " + fileName);
		}
		try
		{
			String lineData = bReader.readLine();
			while ((lineData = bReader.readLine()) != null)
			{
				RowData rowData = getRowData(lineData);
				nowLineNum++;
				cycleCount++;
				if (rowData == null)
				{
					continue;
				}
				makeSQL(rowData.paraId, rowData.equipCode, rowData.paravalue, rowData.localTime);
				if (cycleCount >= LaUDImportTest.testCycleData)
				{
					finishFile = false;
					break;
				}
			}
		}
		catch (IOException e1)
		{
			e1.printStackTrace();
			logger.error(e1.getMessage());
			finishFile = false;
		}
		finally
		{
			if (finishFile)
			{
				logger.info(String.format("import file %s finished!", fileName));
				ProcessManager.commitTask(fileName, true);
				try
				{
					bReader.close();
				}
				catch (IOException e)
				{
					e.printStackTrace();
				}
				catch (NullPointerException e1)
				{
					e1.printStackTrace();
					logger.error(e1.getMessage());
				}
				file = null;
				bReader = null;
				fileName = null;
			}
			else
			{
				ProcessManager.updateFileCount(fileName, 1, nowLineNum);
				ProcessManager.commitTask(fileName, false);
			}
		}
		return true;
	}
	
	public boolean readCSV_JDBC()
	{
		if (file == null)
		{
			file = ProcessManager.getAvailableSource();
			if (file == null)
			{
				return false;
			}
			try
			{
				bReader = new BufferedReader(new FileReader(file));
			}
			catch (FileNotFoundException e3)
			{
				e3.printStackTrace();
				logger.error(e3.getMessage());
				return false;
			}
			fileName = file.getAbsolutePath();
			try
			{
				// 跳过已经导入的
				nowLineNum = getExistingCount();
			}
			catch (IOException e2)
			{
				logger.error("get ExistingCount failed: " + e2.getMessage());
				try
				{
					bReader.close();
				}
				catch (IOException e)
				{
				}
				bReader = null;
				file = null;
				return false;
			}
		}
		// read csv
		boolean success = readData_SQL();
		if (success == false)
		{
			logger.info("Thread read csv failed.");
			return false;
		}
		return true;
	}
	
	private void buildSqlBatch()
	{
		// build sql_batch
		int count = 0;
		StringBuilder sb = new StringBuilder(LaUDTest_JDBC.JDBC_BATCH_SIZE * 100);
		sb.append("BEGIN BATCH USING CONSISTENCY ");
		sb.append(LaUDImportTest.cQLTranscation);
		sb.append("  ");
		for (String sql : sqlList)
		{
			sb.append(sql);
			sb.append(" ");
			count++;
			if (count == LaUDTest_JDBC.JDBC_BATCH_SIZE)
			{
				count = 0;
				sb.append(" APPLY BATCH");
				String sqlString = sb.toString();
				sqlQueue.add(sqlString);
				sb = new StringBuilder(LaUDTest_JDBC.JDBC_BATCH_SIZE * 100);
				sb.append("BEGIN BATCH USING CONSISTENCY ");
				sb.append(LaUDImportTest.cQLTranscation);
				sb.append("  ");
			}
		}
		
		if (count > 0)
		{
			sb.append(" APPLY BATCH");
			String sqlString = sb.toString();
			sqlQueue.add(sqlString);
		}
		
	}
	
	private long getExistingCount() throws IOException
	{
		// 表头
		bReader.readLine();
		long begin = 0;
		ImportProcess process = ProcessManager.getProcessCount(fileName);
		if (process.status == 1)
		{
			begin = process.count;
			for (int k = 0; k < begin; k++)
			{
				bReader.readLine();
			}
		}
		logger.info("get ExistingCount : " + begin);
		return begin;
	}
	
	@Override
	public void run()
	{
		if (readCSV_JDBC())
		{
			buildSqlBatch();
		}
		LaUDImportTest.nowCycleCount += cycleCount;
	}
}
